package com.openess.bigsearch.engine.BPlusTree;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.AbstractCollection;
import java.util.Collection;
import java.util.Iterator;

import com.openess.bigsearch.engine.BTree.util.ByteArrays;
import com.openess.bigsearch.engine.BTree.util.CCollection;
import com.openess.bigsearch.engine.index.FrqPos;
import com.openess.bigsearch.engine.utils.LinkedQueue;

/**
 * 此版本是B树的升级版，B+树实现。和B树的差异：1.关键字和对应的数据全部保存在叶子上。2.叶子的第一个NodeID位置改为保存右兄弟节点的指针，
 * 最左侧的叶子第一个NodeID位置为空
 * 。3.叶子节点中每个关键字对应的数据指针在其右侧。4.叶子节点分裂的时候，分裂的关键字除了升到父节点外，并不从孩子中删除
 * ，而是保存在左孩子中。5.内节点分裂的时候
 * ，分裂方式和B树相同，分裂的关键字不保存在孩子中。6.搜索的时候，如果待搜索关键在内节点中匹配到，那么跟进其左孩子
 * 。7.倒数第二层的节点，其中的key左侧保存小于等于key的孩子
 * ，右侧保存大于key的孩子。8.内节点只是叶子的索引，不带有数据，但叶子点包含所有的关键字，因此内节点中保存的关键字都可以在叶子中找到。
 * 
 * @author TuTu
 * 
 */
public class BPlusTree {
	private RandomAccessFile rf1;
	private RandomAccessFile rf2;
	private FileChannel cf1;
	private FileChannel cf2;
	private File file;
	/**
	 * 树中节点的个数，0代表没有节点
	 */
	private int nodeCount = 0;
	/**
	 * 树中的总key个数，0代表没有
	 */
	private int totalKey = 0;
	/**
	 * 树的高度，因为B树增加高度只会发生在根节点分裂的时候，因此只要根节点分裂的时候，自增即可计数。如果没有任何节点，那么高度为0，只有一个根节点，
	 * 高度为1
	 */
	private int levelHigh;
	/**
	 * 当前插入key对应数据在磁盘中的偏移起始位置
	 */
	private long offset;
	/**
	 * 树的根节点，始终加载到内存中
	 */
	private BTree_node rootNode;
	/**
	 * 存放头信息的字节数组
	 */
	private byte[] head;
	/**
	 * 根节点ID
	 */
	private long rootID;
	/**
	 * 最左端叶子节点的ID号，即保存有所有关键字中按字母顺序最小的一个节点，因为B+树的分裂特性，最左端节点始终不会改变，因此firstID会一直等于0
	 */
	private long firstID;
	/**
	 * 子索引的版本号
	 */
	private int version;
	/**
	 * 关键字大小的比较器
	 */
	private BKComparable compare;

	/**
	 * 头信息占多少字节，头信息包括：1.int存储的nodeCount。2.long存储的rootID。3.long存储的firstID。4.
	 * int存储的levelHigh。5.long存储的下一个待写入data偏移量。6.int存储的叶子关键字个数
	 */
	public static int HEADSIZE = 36;
	/**
	 * 一个节点占多少字节，包括：1.头信息：4个字节的实际key个数，1个字节的leaf判断。2.第一个单独的NodeID或者是sibling指针。3.
	 * <key,ID>对
	 * 
	 */
	public static final int NODESIZE = BTree_node.HEADSIZE + BTree_node.IDSIZE
			+ (BTree_node.MAXKEY) * BTree_node.PAIRESIZE;

	/**
	 * 构造方法，一上来先把根节点加载到内存中，保证根节点始终在内存中，如果磁盘索引是新建的，那么先创建一个根节点
	 * 
	 * @param filePath
	 * @param compare
	 * @param version
	 *            版本号，用于索引合并，0号始终作为主索引
	 * @throws IOException
	 */
	public BPlusTree(String filePath, BKComparable compare, int version)
			throws IOException {
		this.version = version;
		// 关键字索引
		file = new File(filePath + "index_key_" + version);
		rf1 = new RandomAccessFile(file, "rw");
		cf1 = rf1.getChannel();
		// 关键字背后数据的索引
		file = new File(filePath + "index_data_" + version);
		rf2 = new RandomAccessFile(file, "rw");
		cf2 = rf2.getChannel();
		this.compare = compare;
		if (cf1.size() == 0) {
			createRootNode();
		} else {
			init();
		}
	}

	/**
	 * 树的初始化，读取磁盘，将变量赋值
	 */
	private void init() {
		head = readHead();
		nodeCount = ByteArrays.getInt(head, 0);
		rootID = ByteArrays.getLong(head, 4);
		firstID = ByteArrays.getLong(head, 12);
		levelHigh = ByteArrays.getInt(head, 20);
		offset = ByteArrays.getLong(head, 24);
		totalKey = ByteArrays.getInt(head, 32);
		rootNode = getNode(rootID);
	}

	/**
	 * 将头信息从磁盘中读取出来
	 * 
	 * @return 返回读取到的头信息，保存在byte数组中
	 */
	private byte[] readHead() {
		byte[] temp = new byte[HEADSIZE];
		ByteBuffer buf = ByteBuffer.wrap(temp);
		try {
			cf1.read(buf, 0);
		} catch (IOException e) {
			e.printStackTrace();
		}
		return temp;
	}

	/**
	 * 在文件内写入头数据信息，为了支持断点续传，即写入索引中途停止，再次启动程序仍然可以继续写入。我们在头信息所具有的变量发生变化的时候，就自动将其写回
	 * 。集中在节点分裂的时候，包括头分裂和普通分裂。
	 */
	private void writeHead() {
		head = new byte[HEADSIZE];
		ByteArrays.putInt(nodeCount, head, 0);
		ByteArrays.putLong(rootID, head, 4);
		ByteArrays.putLong(firstID, head, 12);
		ByteArrays.putInt(levelHigh, head, 20);
		ByteArrays.putLong(offset, head, 24);
		ByteArrays.putInt(totalKey, head, 32);
		ByteBuffer buf = ByteBuffer.wrap(head);
		try {
			cf1.write(buf, 0);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	/**
	 * 创建根节点
	 */
	private void createRootNode() {
		this.rootNode = new BTree_node(nodeCount++);
		this.rootID = rootNode.nodeID;
		// firstID一旦标记后，就不需要管了，不管节点分裂了多少次，其整棵树的firstID都不会发生改变
		this.firstID = rootNode.nodeID;
		rootNode.leaf = true;
		levelHigh = 1;
	}

	/**
	 * 插入字符类型的关键字
	 * 
	 * @param key
	 */
	public void insert(String str, FrqPos[] data) {
		byte[] strKey = createStrKey(str);
		insertFromRoot(strKey, data);
	}

	/**
	 * 创建一个专门String类型的关键字，前两个字节存储了字符个数，用于以后读取
	 * 
	 * @param str
	 * @return
	 */
	private byte[] createStrKey(String str) {
		return ByteArrays.putString(str);
	}

	/**
	 * 关键字从根节点进行插入。
	 * 关键字插入过程理解：一开始只有一个空的根节点，并且为叶子节点。因为插入操作只会发生在叶子节点上，因此直到根节点被插满，这时候要增加新的节点层
	 * ，第一个非页节点诞生
	 * ，也就是新的根节点。在一颗有N个节点的多层树中，进行插入，同样遵循：只能在叶节点上进行插入。每当叶结点满时，并不是叶节点升为内节点
	 * ，而是叶结点的其中一个值升入内节点
	 * ，并把1个叶子节点拆成2个叶子节点。因此实际上从一开始是叶子的节点，即使插入了N多的key，其最终还是叶结点，永远无法变成内节点
	 * ，叶结点的增加只会发生在分裂时
	 * ，分裂出一个兄弟节点，其也是叶节点。因为所有的插入操作都只发生在叶节点上，因此内结点的key个数增加，就完全依靠其孩子节点分裂一次
	 * ，内节点就增加一个溢出key。
	 * 
	 * 根据这个原理，如何设置节点是否为叶节点，有三个地方：1.在创建新树的根节点的时候leaf=true。2.根节点分裂的时候将新的根leaf=
	 * false。3.普通节点分裂的时候，node.leaf = brotherNode.leaf。
	 * 
	 * @param key
	 * @param data
	 */
	public void insertFromRoot(byte[] key, FrqPos[] data) {
		if (rootNode.isFull()) {
			splitRoot();
		}
		insert(key, rootNode, data);
	}

	/**
	 * 从指定的子树节点开始进行插入操作，这个过程不能单独运行，因为没有进行根节点的分裂操作，运行前提：根节点已进行过分裂。此处没有使用尾递归，
	 * 而是使用循环，可以提高效率。
	 * 
	 * @param key
	 * @param node
	 * 
	 */
	private void insert(byte[] key, BTree_node node, FrqPos[] data) {
		while (true) {
			int result = node.search(key);
			// 如果节点内已有相同的key，那么直接退出
			if (result >= 0)
				return;
			else {
				result = -result - 1;
				// 插入操作只发生在叶子节点上
				if (node.isLeaf()) {
					totalKey++;
					// 顺序不能颠倒，一定要先插入关键字，然后再将key对应数据写入磁盘，因为数据在写入磁盘的过程中，会改变offset的值
					node.insertKeyPaire(key, offset, result);
					BTree_data d = new BTree_data(data);
					d.write();
					// 写回磁盘
					node.write();
					return;
				} else {
					// 将关键字插入这个孩子节点
					BTree_node childNode = node.getChildNode(result);
					// 如果节点中key满，进行分裂
					if (childNode.isFull()) {
						byte[] key1 = splitNode(node, childNode, result);
						// 和分裂后插入父节点的关键字进行比较，如果大于插入的关键字，那么插入位置需要+1，我们采用重新搜索这个节点的方式
						if (compare.compare(key, key1, 0, BTree_node.KEYSIZE) > 0) {
							// 重新搜索当前节点
							childNode = node;
						}
					}
					node = childNode;
				}
			}
		}
	}

	/**
	 * 如果根节点满了，那么需要将根节点进行分裂，并重新设置根节点
	 */
	private void splitRoot() {
		// 新的根节点
		BTree_node newRoot = new BTree_node(nodeCount++);
		// 原先的根节点变成左孩子
		long leftChildID = rootNode.nodeID;
		// 拆分根节点
		byte[] paire = rootNode.splitNode();
		// 往新的根节点内放入数据
		newRoot.insertKeyPaire(paire, 0);
		ByteArrays.putLong(leftChildID, newRoot.data, BTree_node.HEADSIZE);
		// 在内存中替换掉旧的根节点
		setRoot(newRoot);
		// 树的高度加1
		levelHigh++;
		newRoot.write();
		// 根节点发生变化，写回头信息
		writeHead();
	}

	/**
	 * 将节点分裂，并完成将溢出的关键字插入父节点的过程，内含将父节点写回磁盘的操作
	 * 
	 * @param parent
	 *            被分裂节点的父节点
	 * @param node
	 *            被分裂的节点
	 * @param insertIndex
	 *            在父节点的什么位置将溢出值插入
	 * @return 插入到父节点中的关键字
	 */
	private byte[] splitNode(BTree_node parent, BTree_node node, int insertIndex) {
		byte[] paire = node.splitNode();
		parent.insertKeyPaire(paire, insertIndex);
		// 写回磁盘
		parent.write();
		// 节点个数发生变化，写回头信息
		writeHead();
		return paire;
	}

	/**
	 * 将节点设置为树的根节点
	 * 
	 * @param node
	 *            要被设成根节点
	 */
	private void setRoot(BTree_node node) {
		this.rootID = node.nodeID;
		this.rootNode = node;
	}

	/**
	 * 搜索字符类关键字
	 * 
	 * @param str
	 * @return 如果没有搜索到指定的关键字，那么返回空
	 */
	public FrqPos[] searchKey(String str) {
		// 不能使用默认比较器
		compare = new StringBKCompare();
		byte[] strKey = createStrKey(str);
		return searchKeyTree(strKey);
	}

	/**
	 * 将数组中,以字节存储的字符，转换成String类型。为了转换需要先拿到字符的长度。
	 * 
	 * @param strKey
	 * @return
	 */
	public static String readStrKey(byte[] strKey) {
		int length = ByteArrays.getShort(strKey, 0);
		return ByteArrays.getString(strKey, BTree_node.STRLENGTH, length);
	}

	/**
	 * 从整颗树中检索关键字
	 * 
	 * @param key
	 * @return 如果没有搜索到指定的关键字，那么返回空
	 */
	public FrqPos[] searchKeyTree(byte[] key) {
		BTree_data d = searchKey(key, rootNode);
		if (d == null) {
			return null;
		}
		return d.getArray();
	}

	/**
	 * 从指定的子树中检索关键字
	 * 
	 * @param key
	 *            待搜索的关键字
	 * @param node
	 *            搜索子树的根节点
	 * @return 如果没有搜索到指定的关键字，那么返回空
	 */
	private BTree_data searchKey(byte[] key, BTree_node node) {
		if (node == null || key == null) {
			return null;
		}
		int result;
		// 此方法还可以用尾递归来实现，但用循环的效率更高
		while (node != null) {
			result = node.search(key);
			// 不管本节点是否为叶子，先行判断是否找到了关键字
			if (result < 0) {
				// 因为在节点中childID的数量比key多一个，且两者都是从零开始计数，因此当childIndex =
				// 5时，对应的keyIndex=4
				int childIndex = -result - 1;
				// 如果在本节点内没有找到关键字，并且本节点没有孩子了，那么可以说明搜索结束，树中没有指定的关键字
				node = node.getChildNode(childIndex);
			} else {
				// 如果是在叶节点中result>=0那么直接取对应key的数据即可
				if (node.isLeaf()) {
					// 虽然是利用getChildID()方法，但因为叶子节点，原本应该存储ChildID的地方，改为存储data了
					long dataOffset = node.getChildIDorDataOffset(result + 1);
					return getData(dataOffset);
					// 如果不是叶节点，但同时result>=0那么继续进入下一层，因为此处采用的B+树是改进版，因此除了叶子的父节点外，其他内节点都不保存分裂的关键字，所以在这些内节点搜索到对应的关键字的时候，要跟进的是这个关键字左侧的nodeID，而不是右侧的，因为虽然没有在左孩子中保存分裂的关键字，但不管是否保存，跟进左孩子是一定的。
				} else {
					node = node.getChildNode(result);
				}
			}
		}
		return null;
	}

	/**
	 * 从磁盘中拿到节点数据，并初始化节点变量，因为我们所有的操作都是从根节点开始的，不需要从子树开始的操作，因此没必要在节点内保存本节点的NodeID,
	 * 需要哪个节点只需要从根部遍历下来即可
	 * 
	 * @param nodeID
	 * @return 如果节点编号小于0，那么返回空
	 */
	private BTree_node getNode(long nodeID) {
		if (nodeID < 0) {
			return null;
		}
		BTree_node node = new BTree_node(nodeID);
		long nodeOffset = nodeOffset(nodeID);
		ByteBuffer buf = ByteBuffer.wrap(node.data);
		try {
			cf1.read(buf, nodeOffset);
			// 关键字个数
			node.keyCount = ByteArrays.getInt(node.data, 0);
			// 是否为页节点
			node.leaf = ByteArrays.getBoolean(node.data,
					BTree_node.KEYCOUNTSIZE);
		} catch (IOException e) {
			e.printStackTrace();
		}
		return node;
	}

	/**
	 * 根据偏移量，从磁盘中读取出数据，并创建data的实例
	 * 
	 * @param offset
	 * @return
	 */
	public BTree_data getData(long offset) {
		BTree_data d = new BTree_data();
		d.readData(offset);
		return d;
	}

	/**
	 * 在整个磁盘中寻找指定节点的偏移量
	 * 
	 * @param nodeID
	 *            节点的ID
	 * @return 偏移量
	 */
	private long nodeOffset(long nodeID) {
		return HEADSIZE + nodeID * NODESIZE;
	}

	/**
	 * 关闭B树的所有连接
	 * 
	 * @throws IOException
	 */
	public void close() throws IOException {
		this.writeHead();
		this.rootNode.write();
		cf1.close();
		cf2.close();
		rf1.close();
		rf2.close();
	}

	@Override
	// 选用菜单打印方式
	public String toString() {
		return "";
	}

	/**
	 * 菜单方式打印，适合大数据量，并且表现直观
	 * 
	 * @throws IOException
	 */
	public void menuPrint() throws IOException {
		BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(
				new FileOutputStream("E:\\search\\log\\BPlusTree_menu_"
						+ version)));
		try {
			// 打印树的基本信息
			writer.write("BPlusTree, nodeCount: " + nodeCount + ", TotalKey: "
					+ totalKey + ", levelHigh: " + levelHigh + ", rootID: "
					+ rootID + "\n");
			writer.write(rootNode + "\n");
			menuPrint(rootNode, 1, writer);
		} finally {
			if (writer != null)
				writer.close();
		}
	}

	private void menuPrint(BTree_node node, int level, BufferedWriter writer)
			throws IOException {
		String preStr = "";
		for (int i = 0; i < level; i++)
			preStr += "    ";
		int high = level;
		for (int j = 0; j < node.keyCount + 1; j++) {
			BTree_node child = node.getChildNode(j);
			writer.write(preStr + child.toString() + "\n");
			if (!child.leaf)
				menuPrint(child, high + 1, writer);
		}
	}

	/**
	 * 层次打印，数据保存在文本中,仅适合打印小型索引，因为层次打印，会造成一行的数据过多，文本编辑器无法打开
	 * 
	 * @throws IOException
	 */
	public void hierarPrint() throws IOException {
		BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(
				new FileOutputStream("E:\\search\\log\\BPlusTree_hierar_"
						+ version)));
		try {
			// 优先队列，用于层次遍历
			LinkedQueue<BTree_node> que = new LinkedQueue<BTree_node>();
			// 打印树的基本信息
			writer.write("BPlusTree, nodeCount: " + nodeCount + ", TotalKey: "
					+ totalKey + ", levelHigh: " + levelHigh + ", rootID: "
					+ rootID + "\n");
			// 单独打印根节点
			writer.write("level 0 :\n\t" + this.rootNode);
			// 根节点的下一层节点个数，key+1等于孩子节点个数
			int size = rootNode.keyCount + 1;
			// 将下一层的节点全部装入队列
			for (int i = 0; i < size; i++) {
				que.enqueue(rootNode.getChildNode(i));
			}
			// 一层的节点个数
			int count = 0;
			// 从第二层开始打印
			for (int j = 1; j < levelHigh; j++) {
				writer.write("\nlevel " + j + " : \n\t");
				for (int m = 0; m < size; m++) {
					BTree_node p = que.dequeue();
					// 循环遍历当前层所有节点过程中，叠加计算下一层的节点总个数
					count += p.keyCount + 1;
					writer.write(p + " ");
					// 将当前节点的所有孩子装入队列
					for (int n = 0; n < p.keyCount + 1; n++)
						que.enqueue(p.getChildNode(n));
				}
				// 将size设为即将遍历的下一层节点的总个数
				size = count;
				count = 0;
			}
		} finally {
			if (writer != null)
				writer.close();
		}
	}

	/**
	 * 模仿HashMap中的HashIterator类实现方式
	 * 
	 * @author TuTu
	 * 
	 */
	private abstract class AbstractIterator<E> implements Iterator<E> {
		private BTree_node next;
		private long nodeID;

		public AbstractIterator() {
			nodeID = firstID;
		}

		@Override
		public boolean hasNext() {
			return nodeID != -1d;
		}

		public BTree_node nextNode() {
			next = getNode(nodeID);
			nodeID = next.getSiblingID();
			return next;
		}

		@Override
		public void remove() {
		}
	}

	private class NodeIterator extends AbstractIterator<BTree_node> {
		@Override
		public BTree_node next() {
			return nextNode();
		}
	}

	private class TermIterator extends AbstractIterator<String> {
		/**
		 * 记录遍历到当前节点中的第几个key了
		 */
		private int keycount;
		private BTree_node next;

		public TermIterator() {
			// 调用nextNode()已经使得super.hasNext产生了变化，所以必须复写
			next = nextNode();
		}

		@Override
		public boolean hasNext() {
			if (super.hasNext())
				return super.hasNext();
			else
				return keycount < next.keyCount;
		}

		@Override
		public String next() {
			if (keycount >= next.keyCount) {
				next = nextNode();
				keycount = 0;
			}
			byte[] temp = next.getKey(keycount++);
			return readStrKey(temp);
		}
	}

	private class PostIterator extends AbstractIterator<Post> {
		/**
		 * 记录遍历到当前节点中的第几个key了
		 */
		private int keycount;
		private BTree_node next;

		public PostIterator() {
			next = nextNode();
		}

		@Override
		public boolean hasNext() {
			if (super.hasNext())
				return super.hasNext();
			else
				return keycount < next.keyCount;
		}

		@Override
		public Post next() {
			if (keycount >= next.keyCount) {
				next = nextNode();
				keycount = 0;
			}
			byte[] paire = next.getPaire(keycount++);
			byte[] term = ByteArrays.get(paire, 0, BTree_node.KEYSIZE);
			long offset = ByteArrays.getLong(paire, BTree_node.KEYSIZE);
			BTree_data data = getData(offset);
			return new Post(term, data.array);
		}
	}

	/**
	 * BPlusTree中BTree_node的视图
	 * 
	 * @return
	 */
	public Collection<BTree_node> nodes() {
		return new Nodes();
	}

	/**
	 * Btree_node中的<key,dataOffset>视图，其实还是一次性读取一个BTree_node，而不是从磁盘读取单个term
	 * 
	 * @return
	 */
	public Collection<String> terms() {
		return new Terms();
	}

	/**
	 * <term, FrqPos[]>的视图
	 * 
	 * @return
	 */
	public Collection<Post> posts() {
		return new Posts();
	}

	private class Nodes extends AbstractCollection<BTree_node> {
		@Override
		public Iterator<BTree_node> iterator() {
			return new NodeIterator();
		}

		@Override
		public int size() {
			return nodeCount;
		}
	}

	private class Terms extends AbstractCollection<String> {
		@Override
		public Iterator<String> iterator() {
			return new TermIterator();
		}

		@Override
		public int size() {
			return totalKey;
		}
	}

	private class Posts extends AbstractCollection<Post> {
		@Override
		public Iterator<Post> iterator() {
			return new PostIterator();
		}

		@Override
		public int size() {
			return totalKey;
		}
	}

	/**
	 * 倒排列表，<term, FrqPos[]>，现在主要用于索引合并
	 * 
	 * @author xiaoshuo
	 * 
	 */
	public static class Post implements Comparable<Post> {
		private byte[] term;
		private FrqPos[] pos;

		public Post(byte[] term, FrqPos[] pos) {
			this.term = term;
			this.pos = pos;
		}

		@Override
		public int compareTo(Post o) {
			return new StringBKCompare().compare(term, o.term, 0,
					BTree_node.KEYSIZE);
		}

		public byte[] getTerm() {
			return term;
		}

		public FrqPos[] getPos() {
			return pos;
		}

		public String toString() {
			String str = readStrKey(term) + ":";
			for (int i = 0; i < pos.length - 1; i++) {
				str += pos[i] + "->";
			}
			str += pos[pos.length - 1];
			return str;
		}
	}

	/**
	 * B+树的节点
	 * 
	 * @author TuTu
	 * 
	 */
	public class BTree_node implements Comparable<BTree_node> {
		/**
		 * 一个节点内，key的总个数上限，此值对所有节点都是固定的大小
		 */
		public static final int MAXKEY = 70;
		/**
		 * 一个关键字所占字节大小，包含2个字节的字符长度(short)
		 */
		public static final int KEYSIZE = 28;
		/**
		 * 如果关键字是String类型的，那么关键字的前2个字节用于存储字符的个数，用short类型
		 */
		public static final int STRLENGTH = 2;
		/**
		 * 如果关键字是String类型的，那么依据KEYSIZE的大小得出的这个关键字的长度上限
		 */
		public static final int STRSIZE = (KEYSIZE - STRLENGTH) / 2;
		/**
		 * 存储在节点内的一个ID所占的字节数
		 */
		public static final int IDSIZE = 8;
		/**
		 * 叶结点需要存储兄弟节点的指针，一个指针所占的字节数。因为叶节点的第一个ID不像内节点一样存储有数据，
		 * 因此利用这个空间来为叶节点存储兄弟节点的指针
		 */
		public static final int SIBLINGSIZE = IDSIZE;
		/**
		 * 节点内除了存储key,还需要存储childNodeID指针，因此组成(key,childNodeID)，一个k/v所占字节数
		 */
		public static final int PAIRESIZE = KEYSIZE + IDSIZE;
		/**
		 * 头信息中，关键字个数所占的字节数
		 */
		public static final int KEYCOUNTSIZE = 4;
		/**
		 * 头信息中，leaf信息所占的字节数
		 */
		public static final int LEAFSIZE = 1;
		/**
		 * 头信息所占的字节，头信息存储关键字个数,以及一个以byte来存储的boolean值用于判断是否为叶节点
		 */
		public static final int HEADSIZE = KEYCOUNTSIZE + LEAFSIZE;
		/**
		 * 节点的ID号
		 */
		private long nodeID;
		/**
		 * 当前节点内的关键字个数，从0开始，0代表没有关键字，1代表有一个关键字
		 */
		private int keyCount;
		/**
		 * 存放节点内所有数据的数组，最后将数组写入磁盘
		 */
		private byte[] data;
		/**
		 * 判断是否为叶子节点
		 */
		private boolean leaf = false;

		/**
		 * 构造方法，新建一个节点
		 * 
		 * @param nodeID
		 */
		public BTree_node(long nodeID) {
			this.nodeID = nodeID;
			this.data = new byte[NODESIZE];
		}

		/**
		 * 拿到节点中指定关键字的值
		 * 
		 * @param keyIndex
		 *            节点中第几个关键字，从零开始，零代表第一个关键字
		 * @return 关键字值的字节数组
		 */
		public byte[] getKey(int keyIndex) {
			int offset = keyOffset(keyIndex);
			return ByteArrays.get(data, offset, KEYSIZE);
		}

		/**
		 * 拿到节点中指定<key,dataOffset>的值(叶子)
		 * 
		 * @param index
		 *            节点中第几个关键字，从零开始，零代表第一个关键字
		 * @return
		 */
		public byte[] getPaire(int index) {
			int offset = keyOffset(index);
			return ByteArrays.get(data, offset, PAIRESIZE);
		}

		/**
		 * 如果本节点是叶子节点，那么拿到其兄弟节点的NodeID，如果不是叶子节点，那么返回-1
		 * 
		 * @return
		 */
		public long getSiblingID() {
			if (!leaf) {
				return -1;
			}
			long result = ByteArrays.getLong(data, HEADSIZE);
			// 因为最后一个节点没有设置siblingID，拿到的result会等于0，而0是firstID
			if (result == 0d)
				return -1;
			return result;
		}

		/**
		 * 关键字在节点内的偏移量
		 * 
		 * @param keyID
		 *            节点的第几个关键字,从零开始。因为在节点中childID的数量比key多一个，且两者都是从零开始计数，
		 *            因此当childIndex =5时，对应的keyIndex=4
		 * @return
		 */
		private int keyOffset(int keyIndex) {
			return HEADSIZE + IDSIZE + keyIndex * PAIRESIZE;
		}

		/**
		 * 拿到当前节点中保存的孩子节点ID，或者是拿到叶子节点中保存的data偏移量
		 * 
		 * @param childIndex
		 *            孩子节点在本节点中保存位置的编号，不是ID，零代表第一个
		 * @return 如果孩子节点的编号小于0，那么返回-1，代表程序运行错误
		 */
		public long getChildIDorDataOffset(int childIndex) {
			if (childIndex < 0) {
				return -1;
			}
			int offset = childOffset(childIndex);
			long result = ByteArrays.getLong(this.data, offset);
			return result;
		}

		/**
		 * 根据孩子ID存储在本节点中的编号，拿到孩子节点的对象
		 * 
		 * @param childIndex
		 *            孩子节点ID保存在本节点的位置编号
		 * @return 如果孩子节点的编号小于0，或者本节点是叶子节点，那么返回空。同时，如果通过编号找到的nodeID小于0，那么也是返回空。
		 *         这么做出要是为了针对叶节点没有孩子节点的情况
		 */
		private BTree_node getChildNode(int childIndex) {
			if (childIndex < 0 || this.isLeaf()) {
				return null;
			}
			long childID = getChildIDorDataOffset(childIndex);
			BTree_node childNode = getNode(childID);
			return childNode;
		}

		/**
		 * 孩子节点指针在父节点内的偏移量
		 * 
		 * @param childIndex
		 *            此孩子节点指针在父节点内是第几个，从零开始，数量比key的总数多一个
		 * @return
		 */
		private int childOffset(int childIndex) {
			return HEADSIZE + childIndex * PAIRESIZE;
		}

		/**
		 * 仅在本节点内搜索指定的key
		 * 
		 * @param key
		 *            要查找的key
		 * 
		 * @return 
		 *         返回keyIndex，如果返回值大于零，说明查找到指定的关键字。如果返回值小于零，则说明没有找到指定的key，返回值为return
		 *         = -low -1， 根据此值可以计算出应该把关键字插在哪里，只需要反向计算low = -return
		 *         -1，即可拿到low的值
		 *         。根据折半查找，如果没有找到结果，那么折半结束后，low的值即是第一个比key大的值，也就是按顺序插入
		 *         ，key应该插在的地方(keyIndex)
		 */
		public int search(byte[] key) {
			int low = 0;
			int high = keyCount - 1;
			// 如果没有数据，也就不用查找key在data数组中的位置了，直接返回-1，代表放在节点位置0上
			if (isEmpty()) {
				return -1;
			}
			while (low <= high) {
				int mid = (low + high) / 2;
				int midoffset = keyOffset(mid);
				int result = compare.compare(key, data, midoffset, KEYSIZE);
				if (result == 0) {
					return mid;
				} else if (result > 0) {
					low = mid + 1;
				} else {
					high = mid - 1;
				}
			}
			return -low - 1;
		}

		/**
		 * 此方法针对B+树进行了改进。
		 * 当节点内的关键字满了时，进行的分裂操作，只是进行了分裂，并没有将溢出key,nodeID对插入到父节点中，内含将两个节点写回磁盘的操作
		 * 
		 * @return 待被插入到父节点的溢出key,nodeID对
		 */
		public byte[] splitNode() {
			BTree_node newNode = new BTree_node(nodeCount++);
			// 如果当前节点是叶子节点,那么分裂时，1.中间关键字保留在叶子中，2.第一个ID存储siblingID，3.新分裂的节点第一个ID不保存数据，如果不是叶子节点，那么正常分裂
			int interval1 = 0;
			int interval2 = 0;
			if (leaf) {
				interval1 = IDSIZE;
				interval2 = PAIRESIZE;
				// 因为是叶子节点，所以往左侧节点第一个ID位置存入sibling指针
				long oldSibID = getSiblingID();
				ByteArrays.putLong(oldSibID, newNode.data, HEADSIZE);
				ByteArrays.putLong(newNode.nodeID, data, HEADSIZE);
				// 重新设置分裂后两个兄弟节点的关键字个数，叶子和非叶节点稍有不同
				this.keyCount = MAXKEY / 2 + 1;
				newNode.keyCount = MAXKEY - this.keyCount;
			} else {
				// 非叶的情况下，重新设置分裂后两个兄弟节点的关键字个数，非叶节点，有一个关键字升到父节点，叶节点不保存
				this.keyCount = MAXKEY / 2;
				newNode.keyCount = MAXKEY - this.keyCount - 1;
			}
			int splitOffset = keyOffset(MAXKEY / 2);
			// 将分裂后的右半部分数据放入到新节点当中，也因此标有firstID的节点永远不会改变为其他节点
			ByteArrays.put(data, splitOffset + KEYSIZE + interval1,
					newNode.data, HEADSIZE + interval1, NODESIZE - splitOffset
							- KEYSIZE - interval1);
			// 将当前节点的leaf属性传递给分裂的兄弟节点
			newNode.leaf = this.leaf;
			// 拿到中间key
			byte[] midKey = ByteArrays.get(data, splitOffset, KEYSIZE);
			// 组合成key,nodeID对
			byte[] midKeyPaire = createPaire(midKey, newNode.nodeID);
			// 清除当前节点的右半部分
			ByteArrays.clear(data, splitOffset + interval2, NODESIZE
					- splitOffset - interval2);
			// 新创建的节点和本分裂节点都写回磁盘
			newNode.write();
			this.write();
			return midKeyPaire;
		}

		/**
		 * 在节点中插入(key,ID)对,此方法不判断key是否达到上限，应该由使用者(BTree_new)来判断
		 * 
		 * @param key
		 * @param IDorOffset
		 *            可能是nodeID，也可能是dataOffset
		 * @param keyIndex
		 */
		public <E extends CCollection<?>> void insertKeyPaire(byte[] key,
				long IDorOffset, int keyIndex) {
			byte[] paire = createPaire(key, IDorOffset);
			insertKeyPaire(paire, keyIndex);
		}

		/**
		 * 在节点中插入(key,ID)对,此方法不判断key是否达到上限，应该由使用者(BTree_new)来判断
		 * 
		 * @param paire
		 * @param keyIndex
		 *            关键字编号，从0开始算，0是第一个关键字
		 */
		public void insertKeyPaire(byte[] paire, int keyIndex) {
			int keyOffset = keyOffset(keyIndex);
			// 因为不是尾插入所以需要移动数据，相反如果是尾插入，移动数据过程可以跳过
			if (keyIndex < keyCount) {
				int currentSize = keyOffset(keyCount);
				ByteArrays.move(this.data, keyOffset, keyOffset + PAIRESIZE,
						currentSize - keyOffset);
			}
			ByteArrays.put(paire, 0, data, keyOffset, PAIRESIZE);
			this.keyCount++;
		}

		/**
		 * 对于非叶节点，将一个key和nodeID组合成一个paire字节数组。对于叶节点,将一个key和DataID组合成一个paire字节数组。
		 * 
		 * @param key
		 * @param ID
		 *            既可能是nodeID,也可能是DataID
		 * @return 返回paire字节数组
		 */
		private byte[] createPaire(byte[] key, long ID) {
			byte[] paire = new byte[PAIRESIZE];
			ByteArrays.put(key, 0, paire, 0, KEYSIZE);
			ByteArrays.putLong(ID, paire, KEYSIZE);
			return paire;
		}

		/**
		 * 已经使用完此节点，数据不再变化，先将变量写入数组，再将其写回磁盘
		 */
		private void write() {
			// 写入头信息
			ByteArrays.putInt(keyCount, data, 0);
			ByteArrays.putBoolean(leaf, data, KEYCOUNTSIZE);

			long offset = nodeOffset(nodeID);
			ByteBuffer buf = ByteBuffer.wrap(data);
			try {
				cf1.write(buf, offset);
			} catch (IOException e) {
				e.printStackTrace();
			}
		}

		/**
		 * 判断是否为叶子节点
		 * 
		 * @return 叶子节点为true
		 */
		private boolean isLeaf() {
			return leaf;
		}

		/**
		 * 判断节点内key的个数是否已满
		 * 
		 * @return
		 */
		private boolean isFull() {
			return keyCount == MAXKEY;
		}

		/**
		 * 判断节点内，是否保存有关键字
		 * 
		 * @return
		 */
		private boolean isEmpty() {
			return keyCount == 0;
		}

		public String toString() {
			String str = "nodeID(" + nodeID + ") : {";
			for (int i = 0; i < keyCount; i++) {
				str = str + readStrKey(getKey(i)) + "|";
			}
			str += "} ";
			return str;
		}

		@Override
		public int compareTo(BTree_node o) {
			if (this.nodeID - o.nodeID > 0) {
				return 1;
			} else if (this.nodeID - o.nodeID < 0) {
				return -1;
			} else {
				return 0;
			}
		}
	}

	/**
	 * 一个分词对应的倒排链表的包装类，负责将倒排表写入磁盘，或取出磁盘。其实就是对FrqPos[]包装了一层，写入操作大部分也是调用FrqPos,，
	 * 不过经过包装后，让其写入顺序符合B+树词典表的写入要求
	 * 
	 * @author TuTu
	 * 
	 */
	private class BTree_data {
		/**
		 * 头信息：。1.int存储的FrqPos数组长度，主要是为了确定读取时应该读取的字节长度
		 */
		public static final int HEADSIZE = 4;
		/**
		 * 当前数据的起始偏移量
		 */
		private long offset;
		/**
		 * 装载所有data数据
		 */
		private FrqPos[] array;
		/**
		 * 存储头信息
		 */
		private byte[] head;
		/**
		 * 数组的长度
		 */
		private int length;

		/**
		 * 空的默认构造方法
		 */
		public BTree_data() {

		}

		/**
		 * @param data
		 *            待写入索引的数据
		 */
		public BTree_data(FrqPos[] data) {
			if (data == null) {
				// 创建空数组
				this.array = new FrqPos[0];
			} else {
				this.array = data;
			}
			this.length = array.length;
			this.head = new byte[HEADSIZE];
			this.offset = BPlusTree.this.offset;
		}

		/**
		 * 将data写入磁盘
		 */
		public void write() {
			try {
				// 写入头信息
				ByteArrays.putInt(length, head, 0);
				ByteBuffer buf = ByteBuffer.wrap(head);
				cf2.write(buf, offset);
				offset += BTree_data.HEADSIZE;
				// 写入FrqPos对象
				for (FrqPos frqpos : array) {
					frqpos.write(cf2, offset);
					// offset的值设为下一个data的起始位置
					offset += FrqPos.HEADSIZE + frqpos.getSize();
				}
				// 将偏移量赋值回BPlusTree对象，用于下一次写BTree_data
				BPlusTree.this.offset = offset;
			} catch (IOException e) {
				e.printStackTrace();
			}
		}

		/**
		 * 根据偏移量读取磁盘中的数据
		 * 
		 * @param offset
		 */
		public void readData(long offset) {
			readHead(offset);
			// 接下来读取FrqPos
			offset += HEADSIZE;
			try {
				// 给array变量赋值
				this.array = new FrqPos[length];
				for (int i = 0; i < length; i++) {
					FrqPos frqpos = new FrqPos();
					array[i] = frqpos.read(cf2, offset);
					offset += (FrqPos.HEADSIZE + array[i].getSize());
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
		}

		/**
		 * 将头信息从磁盘中读取出来,并将length,offset三个变量进行初始化，之后才能根据头信息将数据整体读出
		 * 
		 * @param offset
		 */
		private void readHead(long offset) {
			byte[] head = new byte[HEADSIZE];
			try {
				cf2.read(ByteBuffer.wrap(head), offset);
			} catch (IOException e) {
				e.printStackTrace();
			}
			// 读取头信息后，初始化变量
			this.length = ByteArrays.getInt(head, 0);
			this.offset = offset;
		}

		/**
		 * 拿到Data里的数据数组
		 * 
		 * @return
		 */
		public FrqPos[] getArray() {
			return array;
		}

		@Override
		public String toString() {
			String str = "Data(" + offset + ") : {";
			for (int i = 0; i < array.length - 1; i++) {
				str += array[i] + "-";
			}
			return str + array[array.length] + "}";
		}
	}
}
